package cn.edu.flink.scala.tutorial.state.keyedstate

import cn.edu.flink.scala.tutorial.source.Event
import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object ValueStateTest2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    clickStream
      .keyBy(_.user)
      .process(new KeyedProcessFunction[String, Event, String] {
        lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count", classOf[Long]))
        // 定时器时间戳
        lazy val timeTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))


        override def processElement(value: Event, ctx: KeyedProcessFunction[String, Event, String]#Context, out: Collector[String]): Unit = {
          val count = countState.value()
          countState.update(count + 1)

          // 注册定时器
          if (timeTsState.value() == 0L) {
            ctx.timerService().registerEventTimeTimer(value.timestamp + 10 * 1000L)
            // 定时器状态保存
            timeTsState.update(value.timestamp + 10 * 1000L)
          }
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, Event, String]#OnTimerContext, out: Collector[String]): Unit = {
          out.collect(s"用户： ${ctx.getCurrentKey} 的pv是： ${countState.value()}")

          timeTsState.clear()
        }
      })
      .print()


    env.execute("ValueStateTest2")
  }
}
